package juc.executors;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class ComplexExample {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //test1();
        //test2();
        //test3();
        test4();

    }

    private static void test1() throws ExecutionException, InterruptedException {
        final ExecutorService service = Executors.newFixedThreadPool(5);
        //IntStream.range(0,5).boxed().map(integer -> toTask(integer));
        List<Runnable> tasks = IntStream.range(0, 5).boxed().map(ComplexExample::toTask).collect(Collectors.toList());
        List<Future<?>> futureList = new ArrayList<>();
        tasks.forEach(r -> futureList.add(service.submit(r)));
        //future缺点是 等任务全部结束后才get
        futureList.get(4).get();
        futureList.get(3).get();
        futureList.get(2).get();
        futureList.get(1).get();
        futureList.get(0).get();
    }

    private static void test2() throws ExecutionException, InterruptedException {
        final ExecutorService service = Executors.newFixedThreadPool(5);
        List<Runnable> tasks = IntStream.range(0, 5).boxed().map(ComplexExample::toTask).collect(Collectors.toList());
        final CompletionService<Object> completionService = new ExecutorCompletionService<>(service);
        //submit是异步
        //Executors.callable(r)把Runnable转成callback
        tasks.forEach(r -> completionService.submit(Executors.callable(r)));
        Future<?> future;
        while ((future = completionService.take()) != null) {
            System.out.println(future.get());
        }
    }

    private static void test3() throws ExecutionException, InterruptedException {
        final ExecutorService service = Executors.newSingleThreadExecutor();
        List<Runnable> tasks = IntStream.range(0, 5).boxed().map(ComplexExample::toTask).collect(Collectors.toList());
        final CompletionService<Object> completionService = new ExecutorCompletionService<>(service);
        //submit是异步
        //Executors.callable(r)把Runnable转成callback
        tasks.forEach(r -> completionService.submit(Executors.callable(r)));
        TimeUnit.SECONDS.sleep(4);
        //trap，ExecutorService陷阱，返回的是未执行的，执行过程中被中断的不会放入list
        List<Runnable> runnables = service.shutdownNow();
        /**
         * ExecutorCompletionService$QueueingFuture是私有内部类
         * 拿不出没执行完的任务，the task [1] be interrupted
         * CompletionService关注的是完成的任务，不关注未完成的任务
         *
         */
        System.out.println(runnables);
    }

    private static void test4() throws ExecutionException, InterruptedException {
        final ExecutorService service = Executors.newSingleThreadExecutor();
        List<Callable<Integer>> tasks = IntStream.range(0, 5).boxed().map(MyTask::new).collect(Collectors.toList());
        final CompletionService<Integer> completionService = new ExecutorCompletionService<>(service);
        tasks.forEach(completionService::submit);
        TimeUnit.SECONDS.sleep(7);
        List<Runnable> runnables = service.shutdownNow();

        tasks.stream().filter(callable -> !((MyTask) callable).isSuccess())
                .forEach(System.out::println);

    }


    private static class MyTask implements Callable<Integer> {
        private final Integer value;
        private boolean success = false;

        private MyTask(Integer value) {
            this.value = value;
        }

        @Override
        public Integer call() throws Exception {
            System.out.printf("the task [%d] will be executed\n", value);
            TimeUnit.SECONDS.sleep(value + 2);
            System.out.printf("the task [%d] execute done\n", value);
            success = true;
            return value;
        }

        public boolean isSuccess() {
            return success;
        }
    }

    private static Runnable toTask(int i) {
        return () -> {
            try {
                System.out.printf("the task [%d] will be executed\n", i);
                TimeUnit.SECONDS.sleep(i + 2);
                System.out.printf("the task [%d] execute done\n", i);
            } catch (InterruptedException e) {
                System.out.printf("the task [%d] be interrupted\n", i);
            }
        };
    }

    private static void timeSleep(long time) {
        try {
            TimeUnit.SECONDS.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
